package queueAmqp

import (
	"fmt"
	"github.com/spf13/viper"
	"github.com/streadway/amqp"
	"log"
)

type AmqpConfig struct {
	queue        string
	exchangeName string
	exchange     string
	exchangeType string
	routingKey   string
	reliable     bool
	consumerTag  string
}

type Consumer struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	tag     string
}

// 获取amqp连接地址
func getAmqpUrl() (amqpURI string) {

	host := viper.GetString("amqp.host")
	port := viper.GetString("amqp.port")
	username := viper.GetString("amqp.user")
	password := viper.GetString("amqp.password")

	amqpURI = fmt.Sprintf("amqp://%s:%s@%s:%s",
		username,
		password,
		host,
		port)

	return
}

// 通常情况下，一个发布通道、一个序列号和一组未确认的序列号和循环，直到发布通道关闭。
func confirmOne(confirms <-chan amqp.Confirmation) {
	log.Printf("waiting for confirmation of one publishing")

	if confirmed := <-confirms; confirmed.Ack {
		log.Printf("confirmed delivery with delivery tag: %d", confirmed.DeliveryTag)
	} else {
		log.Printf("failed delivery of delivery tag: %d", confirmed.DeliveryTag)
	}
}

// 推送消息
func publishParam(amqpURI, exchange, exchangeType, routingKey string, body []byte, reliable bool) error {

	connection, err := amqp.Dial(amqpURI)
	if err != nil {
		return fmt.Errorf("Dial: %s", err)
	}
	defer connection.Close()

	channel, err := connection.Channel()
	if err != nil {
		return fmt.Errorf("Channel: %s", err)
	}

	if err := channel.ExchangeDeclare(
		exchange,     // name
		exchangeType, // type
		true,         // durable
		false,        // auto-deleted
		false,        // internal
		false,        // noWait
		nil,          // arguments
	); err != nil {
		return fmt.Errorf("Exchange Declare: %s", err)
	}

	// 可靠发布者确认需要确认。选择连接的支持。
	if reliable {
		if err := channel.Confirm(false); err != nil {
			return fmt.Errorf("Channel could not be put into confirm mode: %s", err)
		}
		confirms := channel.NotifyPublish(make(chan amqp.Confirmation, 1))
		defer confirmOne(confirms)
	}

	if err = channel.Publish(
		exchange,   // publish to an exchange
		routingKey, // routing to 0 or more queues
		false,      // mandatory
		false,      // immediate
		amqp.Publishing{
			Headers:         amqp.Table{},
			ContentType:     "text/plain",
			ContentEncoding: "",
			Body:            body,
			DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent
			Priority:        0,              // 0-9
			// a bunch of application/implementation-specific fields
		},
	); err != nil {
		return fmt.Errorf("Exchange Publish: %s", err)
	}

	return err
}

// 实例化消费者
func NewConsumer(amqpURI, exchange, exchangeType, queueName, key, ctag string, handle func(<-chan amqp.Delivery)) {
	c := &Consumer{
		conn:    nil,
		channel: nil,
		tag:     ctag,
	}

	var err error

	c.conn, err = amqp.Dial(amqpURI)
	if err != nil {
		fmt.Printf("Dial: %s", err)
	}

	go func() {
		fmt.Printf("closing: %s", <-c.conn.NotifyClose(make(chan *amqp.Error)))
	}()

	c.channel, err = c.conn.Channel()
	if err != nil {
		fmt.Printf("Channel: %s", err)
	}

	if err = c.channel.ExchangeDeclare(
		exchange,     // name of the exchange
		exchangeType, // type
		true,         // durable
		false,        // delete when complete
		false,        // internal
		false,        // noWait
		nil,          // arguments
	); err != nil {
		fmt.Printf("Exchange Declare: %s", err)
	}

	queue, err := c.channel.QueueDeclare(
		queueName, // name of the queue
		true,      // durable
		false,     // delete when unused
		false,     // exclusive
		false,     // noWait
		nil,       // arguments
	)
	if err != nil {
		fmt.Printf("Queue Declare: %s", err)
	}

	if err = c.channel.QueueBind(
		queue.Name, // name of the queue
		key,        // bindingKey
		exchange,   // sourceExchange
		false,      // noWait
		nil,        // arguments
	); err != nil {
		fmt.Printf("Queue Bind: %s", err)
	}

	deliveries, err := c.channel.Consume(
		queue.Name, // name
		c.tag,      // consumerTag,
		false,      // noAck
		false,      // exclusive
		false,      // noLocal
		false,      // noWait
		nil,        // arguments
	)
	if err != nil {
		fmt.Printf("Queue Consume: %s", err)
	}

	forever := make(chan bool)
	handle(deliveries)
	<-forever
}
